package cn.jly.flink.sideoutput;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.ArrayList;
import java.util.List;

/**
 * 分流场景
 * 我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。
 * 面对这样的需求该如何操作呢？
 *
 * @author lanyangji
 * @create 2020-09-03 17:08
 */
public class SideOutputApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源
        List<Tuple3<Integer, Integer, Integer>> data = new ArrayList<>();
        data.add(new Tuple3<>(0, 1, 0));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(0, 1, 3));
        data.add(new Tuple3<>(1, 2, 5));
        data.add(new Tuple3<>(1, 2, 9));
        data.add(new Tuple3<>(1, 2, 11));
        data.add(new Tuple3<>(1, 2, 13));


        DataStreamSource<Tuple3<Integer, Integer, Integer>> items = env.fromCollection(data);

        // outputTag
        OutputTag<Tuple3<Integer, Integer, Integer>> zeroStreamOutputTag = new OutputTag<Tuple3<Integer, Integer, Integer>>("zeroStream"){};
        OutputTag<Tuple3<Integer, Integer, Integer>> oneStreamOutputTag = new OutputTag<Tuple3<Integer, Integer, Integer>>("oneStream"){};

        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> process = items.process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception {
                if (value.f0 == 0) {
                    ctx.output(zeroStreamOutputTag, value);
                } else if (value.f0 == 1) {
                    ctx.output(oneStreamOutputTag, value);
                }
            }
        });

        // 获取侧输出流
        DataStream<Tuple3<Integer, Integer, Integer>> zeroDataStream = process.getSideOutput(zeroStreamOutputTag);
        DataStream<Tuple3<Integer, Integer, Integer>> oneDataStream = process.getSideOutput(oneStreamOutputTag);

        zeroDataStream.printToErr("zero");
        oneDataStream.printToErr("one");

        env.execute("SideOutputApp");
    }
}
